package com.shujia.flink.cep

import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._

object Demo1Cep {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 1,192.168.0.1,a
      * 1,192.168.0.2,b
      * 1,192.168.0.3,c
      * 1,192.168.0.4,d
      *
      *
      *
      * 匹配数据流中  a之后出现了b
      *
      *
      * 匹配a 出现之后出现了多个b  (2,4)
      *
      */


    val ds: DataStream[String] = env.socketTextStream("master", 8888)

    val events: DataStream[Event] = ds.map(line => {
      val split: Array[String] = line.split(",")
      Event(split(0), split(1), split(2))
    })


    /**
      * 定义匹配模式
      *
      */

    val pattern: Pattern[Event, Event] = Pattern
      .begin[Event]("begin")
      .where(evrnt => evrnt.t.equals("a"))
      .notNext("next")
      .where(event => event.t.equals("b"))



    //对数据流进行匹配

    val pDS: PatternStream[Event] = CEP.pattern(events.keyBy(_.id), pattern)


    //取出匹配成功的数据

    val resultDS = pDS.select(m => m)


    //打印结果
    resultDS.print()


    env.execute()

  }

  case class Event(id: String, ip: String, t: String)

}
